﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Collections.Concurrent;
using System.Threading.Channels;

namespace CTP.CRM.Core.Rabbit
{
    public abstract class RabbitClient
    {
        private readonly ILogger _logger;

        protected readonly RabbitOptions RabbitOptions;
        protected ConcurrentDictionary<IConnection, int> Connections = new();
        protected static ConcurrentDictionary<string, Lazy<IChannel>> Channels = new();

        public RabbitClient(ILogger<RabbitClient> logger, RabbitOptions rabbitOptions)
        {
            _logger = logger;
            RabbitOptions = rabbitOptions;

            var cf = new ConnectionFactory
            {
                AutomaticRecoveryEnabled = true,
                NetworkRecoveryInterval = TimeSpan.FromSeconds(30),
                // see：https://github.com/rabbitmq/rabbitmq-dotnet-client/issues/1112

                Uri = new Uri(RabbitOptions.ConnectionString)
            };
            IConnection conn = cf.CreateConnectionAsync().GetAwaiter().GetResult();
            Connections.TryAdd(conn, 0);
        }

        public async Task Start()
        {
            if ((RabbitOptions.Declarations?.Count ?? 0) > 0)
            {
                using var chnl = await GetConnection().CreateChannelAsync();
                var decl = new DefaultDeclaration(chnl);
                RabbitOptions.Declarations?.ForEach(q => q.Invoke(decl));
            }

            if ((RabbitOptions.Consumes?.Count ?? 0) > 0)
            {
                foreach (var opt in RabbitOptions.Consumes)
                {
                    await Subscribe(opt);
                }
            }
        }

        public async Task Subscribe(ConsumeOptions consumeOptions)
        {
            var channel = await CreateAndConfigureChannel(consumeOptions);
            _ = Channels.GetOrAdd(consumeOptions.Queue, new Lazy<IChannel>(channel));
        }

        private async Task<IChannel> CreateAndConfigureChannel(ConsumeOptions consumeOptions)
        {
            var opt = consumeOptions;
            var channel = await GetConnection().CreateChannelAsync();
            var decl = new DefaultDeclaration(channel);
            opt.Declaration?.Invoke(decl);

            if (opt.FetchCount > 0)
            {
                await channel.BasicQosAsync(0, opt.FetchCount, false);
            }

            var consumer = new AsyncEventingBasicConsumer(channel);

            consumer.ReceivedAsync += (obj, arg) =>
            {
                _logger.LogInformation($"rabbit on queue({opt.Queue}) received msg");

                _ = Task.Run(async () =>
                {
                    try
                    {
                        await OnReceived(obj, arg, opt);
                        await channel.BasicAckAsync(arg.DeliveryTag, false);
                    }
                    catch (Exception ex)
                    {
                        _logger.LogError($"rabbit on queue({opt.Queue}) received error: {ex}");
                        await channel.BasicNackAsync(arg.DeliveryTag, false, opt.FailedRequeue);
                    }
                });

                return Task.CompletedTask;
            };


            // 当通道调用的回调中发生异常时发出信号
            channel.CallbackExceptionAsync += (_, args) => { _logger.LogError(args.Exception, args.Exception.Message); return Task.CompletedTask; };
            await channel.BasicConsumeAsync(
                 queue: opt.Queue,
                 autoAck: true,
                 consumer: consumer);

            return channel;
        }

        public async Task UnSubscribe(string queue, string exchange, string routingKey, bool deleteQueue = false)
        {
            if (!Channels.TryGetValue(queue, out var channel))
            {
                return;
            }

            _logger.LogInformation($"尝试解除rabbit绑定 queue:{queue}, exchange:{exchange}, routingKey:{routingKey}");
            await channel.Value.QueueUnbindAsync(queue, exchange, routingKey);

            if (deleteQueue)
            {
                var passive = await channel.Value.QueueDeclarePassiveAsync(queue);
                _logger.LogInformation(
                    $"尝试删除rabbit queue:{passive.QueueName} consumerCount:{passive.ConsumerCount} messageCount:{passive.MessageCount}");
                await channel.Value.QueueDeleteAsync(queue);
            }

            await channel.Value.CloseAsync();
            Channels.TryRemove(queue, out _);
        }

        public async Task Stop()
        {
            if (Channels != null)
            {
                foreach (var kv in Channels)
                {
                    await kv.Value.Value.CloseAsync();
                }

                Channels = null;
            }

            if (Connections != null)
            {
                foreach (var kv in Connections)
                {
                    await kv.Key.CloseAsync();
                }

                Connections = null;
            }
        }

        public async Task Publish(string exchange, string routingKey, byte[] data, Action<IBasicProperties> options = null)
        {
            using var activity = RabbitInstrumentation.ActivitySource.StartActivity($"/{exchange}/{routingKey}");
            activity?.SetTag("kind", "publish");

            var key = $"{exchange}:{routingKey}";

            var chnl = await GetConnection().CreateChannelAsync();
            var channel = Channels.GetOrAdd(key, _ => new Lazy<IChannel>(chnl));

            //var prop = channel.Value.CreateBasicProperties();
            //options?.Invoke(prop);

            var prop = new BasicProperties();
            options?.Invoke(prop);
         
            await channel.Value.BasicPublishAsync(exchange: exchange, routingKey: routingKey, mandatory: true, basicProperties: prop, body: data);
        }

        protected abstract Task OnReceived(object model, BasicDeliverEventArgs args, ConsumeOptions options);

        private IConnection GetConnection()
        {
            var con = Connections.OrderBy(q => q.Value).Select(q => q.Key).FirstOrDefault();
            Connections.AddOrUpdate(con, _ => 1, (_, v) => v + 1);
            return con;
        }

    }
}
